/*
 * Copyright (c) 2020-2099 the original author or authors. All rights reserve.
 */
package org.tianyun.cloud.data.redis.factory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.tianyun.cloud.utils.JsonUtils;

import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.Type;
import java.math.BigDecimal;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * redis操作工具类
 *
 * @auther ebert_chan
 */
public class ObjectRedisTemplate extends RedisTemplate<String, byte[]> implements Closeable {

    private static final Logger LOGGER = LoggerFactory.getLogger(ObjectRedisTemplate.class);

    private static final RedisScript<Long> SCRIPT_INCREMENT = RedisScript.of("local result=redis.call('incrBy', KEYS[1], ARGV[1]) if result == 1 then redis.call('expire', KEYS[1], ARGV[2]) end return result", Long.class);

    private static final RedisScript<Long> SCRIPT_UNLOCK = RedisScript.of("local result=redis.call('get', KEYS[1]) if result == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", Long.class);

    private final RedisSerializer<Object> jdkRedisSerializer;

    private final RedisSerializer<byte[]> byteArrayRedisSerializer;

    private final RedisSerializer<String> stringRedisSerializer;

    private boolean destoryFlag = false;

    public ObjectRedisTemplate() {
        this.jdkRedisSerializer = RedisSerializer.java();
        this.byteArrayRedisSerializer = RedisSerializer.byteArray();
        this.stringRedisSerializer = RedisSerializer.string();

        setKeySerializer(this.stringRedisSerializer);
        setValueSerializer(this.byteArrayRedisSerializer);
        setHashKeySerializer(this.stringRedisSerializer);
        setHashValueSerializer(this.byteArrayRedisSerializer);
    }

    /**
     * 设置key的值，value可以是基本数据类型或者实现java.io.Serializable接口，如果都不是会将对象序列化为json格式字符串， 请确保被序列化为json后能够被反序列化。
     *
     * @param key   键
     * @param value 值
     * @author ebert_chan
     */
    public void set(String key, Object value) {
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        this.opsForValue().set(key, bytes);
    }

    /**
     * 设置key的值，value可以是基本数据类型或者实现java.io.Serializable接口，如果都不是会将对象序列化为json格式字符串， 请确保被序列化为json后能够被反序列化。
     *
     * @param key     键
     * @param value   值
     * @param timeout 过期时间，单位秒
     * @auther ebert_chan
     */
    public void set(String key, Object value, int timeout) {
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        this.opsForValue().set(key, bytes, timeout);
    }

    /**
     * 设置key的值，value可以是基本数据类型或者实现java.io.Serializable接口，如果都不是会将对象序列化为json格式字符串， 请确保被序列化为json后能够被反序列化。
     *
     * @param key     键
     * @param value   值
     * @param timeout 过期时间
     * @param unit    时间单位
     * @auther ebert_chan
     */
    public void set(String key, Object value, long timeout, TimeUnit unit) {
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        this.opsForValue().set(key, bytes, timeout, unit);
    }

    /**
     * 只有键key不存在的时候才会设置key的值，value可以是基本数据类型或者实现java.io.Serializable接口， 如果都不是会将对象序列化为json格式字符串， 请确保被序列化为json后能够被反序列化。
     *
     * @param key   键
     * @param value 值
     * @return
     * @auther ebert_chan
     */
    public boolean setIfAbsent(String key, Object value) {
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        Boolean isOk = this.opsForValue().setIfAbsent(key, bytes);
        return isOk != null && isOk;
    }

    /**
     * 只有键key不存在的时候才会设置key的值，value可以是基本数据类型或者实现java.io.Serializable接口， 如果都不是会将对象序列化为json格式字符串， 请确保被序列化为json后能够被反序列化。
     *
     * @param key     键
     * @param value   值
     * @param timeout 过期时间，单位秒
     * @return
     * @auther ebert_chan
     */
    public boolean setIfAbsent(String key, Object value, int timeout) {
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        Boolean isOk = this.opsForValue().setIfAbsent(key, bytes, timeout, TimeUnit.SECONDS);
        return isOk != null && isOk;
    }

    /**
     * 只有键key存在的时候才会设置key的值，value可以是基本数据类型或者实现java.io.Serializable接口， 如果都不是会将对象序列化为json格式字符串， 请确保被序列化为json后能够被反序列化。
     *
     * @param key   键
     * @param value 值
     * @return
     * @auther ebert_chan
     */
    public boolean setIfPresent(String key, Object value) {
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        Boolean isOk = this.opsForValue().setIfPresent(key, bytes);
        return isOk != null && isOk;
    }

    /**
     * 只有键key存在的时候才会设置key的值，value可以是基本数据类型或者实现java.io.Serializable接口， 如果都不是会将对象序列化为json格式字符串， 请确保被序列化为json后能够被反序列化。
     *
     * @param key     键
     * @param value   值
     * @param timeout 过期时间，单位秒
     * @return
     * @auther ebert_chan
     */
    public boolean setIfPresent(String key, Object value, int timeout) {
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        Boolean isOk = this.opsForValue().setIfPresent(key, bytes, timeout, TimeUnit.SECONDS);
        return isOk != null && isOk;
    }

    /**
     * 自增长
     *
     * @param key     键
     * @param timeout 过期时间，单位秒
     * @return
     * @auther ebert_chan
     */
    public Long increment(String key, int timeout) {
        List<String> keys = Collections.singletonList(key);
        return this.execute(SCRIPT_INCREMENT, keys, "1".getBytes(), String.valueOf(timeout).getBytes());
    }

    /**
     * 尝试创建分布式锁，成功返回true
     * <p>
     *
     * @param key   加锁key值
     * @param value 该值用于确定加锁者身份
     * @return
     * @author ebert_chan
     * @date 2018年11月2日
     */
    public boolean tryLock(String key, String value) {
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        Boolean isLock = this.opsForValue().setIfAbsent(key, bytes);
        return isLock != null && isLock;
    }

    /**
     * 尝试创建分布式锁，锁会在milliseconds毫秒之后自动释放，成功返回true
     *
     * @param key          加锁key值
     * @param value        该值用于确定加锁者身份
     * @param milliseconds 过期时间，单位毫秒
     * @return
     * @author ebert_chan
     */
    public boolean tryLock(String key, String value, long milliseconds) {
        if (milliseconds <= 0) {
            return this.tryLock(key, value);
        }
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        Boolean isLock = this.opsForValue().setIfAbsent(key, bytes, milliseconds, TimeUnit.MILLISECONDS);
        return isLock != null && isLock;
    }

    /**
     * 尝试创建分布式锁，如获取锁失败，进入等待，直至waitTime后还未获取锁，则返回false； 如获取锁成功则seconds秒之后自动释放锁
     *
     * @param key          加锁key值
     * @param value        该值用于确定加锁者身份
     * @param milliseconds 过期时间，单位毫秒
     * @param waitTime     等待时间，单位毫秒
     * @return
     * @author ebert_chan
     */
    public boolean tryLock(String key, String value, long milliseconds, long waitTime) {
        if (waitTime <= 0) {
            return this.tryLock(key, value, milliseconds);
        }
        Boolean isLock = false;
        long millis = Math.min(60, waitTime);
        boolean loop;
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));

        try {
            do {
                isLock = this.opsForValue().setIfAbsent(key, bytes, milliseconds, TimeUnit.MILLISECONDS);
                loop = (isLock == null || !isLock) && waitTime > 0;
                if (loop && !destoryFlag) {
                    Thread.sleep(Math.min(millis, waitTime));
                }
                waitTime -= millis;
            } while ((isLock == null || !isLock) && !destoryFlag && loop);
        } catch (Exception e) {
            LOGGER.warn("Get redis distributed lock exception, key={}", key, e);
        }
        return isLock != null && isLock;
    }

    /**
     * 释放锁
     * <p>
     *
     * @param key
     * @param value
     * @return
     * @author ebert_chan
     */
    public boolean unlock(String key, String value) {
        List<String> keys = Collections.singletonList(key);
        byte[] bytes = this.defaultEmptyBytes(this.serializeToBytes(value));
        Long result = this.execute(SCRIPT_UNLOCK, keys, new Object[]{bytes});
        return result != null && result > 0;
    }

    /**
     * 从缓存获取值，如果明确redis中的数据可以反序列化为string类型，则可以使用此方法，否则可能会乱码
     *
     * @param key
     * @return
     * @auther ebert_chan
     */
    public String get(String key) {
        return this.get(key, String.class);
    }

    /**
     * 从缓存获取值，反序列化为指定类型的对象
     * <p>
     *
     * @param key       缓存key
     * @param valueType 数据类型
     * @return
     * @author ebert_chan
     */
    public <T> T get(String key, Class<T> valueType) {
        byte[] bytes = this.opsForValue().get(key);
        T value = null;
        try {
            value = this.deserializeForBytes(bytes, valueType);
        } catch (Throwable e) {
            LOGGER.warn("Deserialize to object failed. Cause by: {}", e.getMessage());
            this.delete(key);
        }
        return value;
    }

    /**
     * 从缓存获取值，如果缓存不存在，调用指定方法获取结果后自动设置缓存。序列化方法：基础数据类型返回字符串；List、Map类型使用JDK序列化方式；其他类型默认序列化为JSON格式字符串
     * <p>
     *
     * @param valueMapper 获取数据方法
     * @param key         缓存key
     * @param timeout     过期时间
     * @param unit        时间单位
     * @param valueType   数据类型
     * @return
     * @author ebert_chan
     */
    public <T> T getAndSet(Supplier<T> valueMapper, String key, long timeout, TimeUnit unit, Class<T> valueType) {
        byte[] bytes = this.opsForValue().get(key);
        T t;
        if (bytes == null) {
            t = valueMapper.get();
            this.set(key, t, timeout, unit);
        } else {
            try {
                t = this.deserializeForBytes(bytes, valueType);
            } catch (Throwable e) {
                LOGGER.error("Deserialize to object failed. Cause by:", e);
                this.delete(key);
                return null;
            }
        }
        return t;
    }

    /**
     * 从缓存获取值，如果缓存不存在，调用指定方法获取结果后自动设置缓存。序列化方法：基础数据类型返回字符串；List、Map类型使用JDK序列化方式；其他类型默认序列化为JSON格式字符串
     * <p>
     *
     * @param valueMapper 获取数据方法
     * @param key         缓存key
     * @param timeout     过期时间，单位秒
     * @param valueType   数据类型
     * @return
     * @author ebert_chan
     */
    public <T> T getAndSet(Supplier<T> valueMapper, String key, long timeout, Class<T> valueType) {
        return this.getAndSet(valueMapper, key, timeout, TimeUnit.SECONDS, valueType);
    }

    /**
     * 从缓存获取值，如果缓存不存在，调用指定方法获取结果后自动设置缓存
     * <p>
     *
     * @param valueMapper 获取数据方法
     * @param key         缓存key
     * @param timeout     过期时间，单位秒
     * @return
     * @author ebert_chan
     */
    public String getAndSet(Supplier<String> valueMapper, String key, long timeout) {
        return getAndSet(valueMapper, key, timeout, String.class);
    }

    /**
     * 从缓存获取值，如果缓存不存在，调用指定方法获取结果后自动设置缓存。序列化方法：基础数据类型返回字符串；其他类型默认序列化为JSON格式字符串
     * <p>
     *
     * @param valueMapper   获取数据方法
     * @param key           缓存key
     * @param timeout       过期时间
     * @param unit          时间单位
     * @param typeReference 数据类型
     * @return
     * @author ebert_chan
     */
    public <T> T getAndSet(Supplier<T> valueMapper, String key, long timeout, TimeUnit unit, TypeReference<T> typeReference) {
        byte[] bytes = this.opsForValue().get(key);
        T t;
        if (bytes == null) {
            t = valueMapper.get();
            bytes = this.defaultEmptyBytes(this.serializeToString(t));
            this.set(key, bytes, timeout, unit);
        } else {
            try {
                if (bytes.length == 0) {
                    return null;
                }
                t = this.deserializeForStringValue(stringRedisSerializer.deserialize(bytes), typeReference);
            } catch (Throwable e) {
                LOGGER.error("Deserialize to object failed. Cause by:", e);
                this.delete(key);
                return null;
            }
        }
        return t;
    }

    /**
     * 从缓存获取值，如果缓存不存在，调用指定方法获取结果后自动设置缓存。序列化方法：基础数据类型返回字符串；其他类型默认序列化为JSON格式字符串
     * <p>
     *
     * @param valueMapper   获取数据方法
     * @param key           缓存key
     * @param timeout       过期时间，单位秒
     * @param typeReference 数据类型
     * @return
     * @author ebert_chan
     */
    public <T> T getAndSet(Supplier<T> valueMapper, String key, long timeout, TypeReference<T> typeReference) {
        return this.getAndSet(valueMapper, key, timeout, TimeUnit.SECONDS, typeReference);
    }

    /**
     * 从缓存批量获取数据。如果缓存不存在数据，汇集keys的key来批量获取数据后追条设置缓存。序列化方法：基础数据类型返回字符串；其他类型默认序列化为JSON格式字符串
     * <p>
     *
     * @param keys        缓存keys，key=T，value=缓存key
     * @param valueMapper 获取数据方法，通过keys批量获取数据
     * @param timeout     过期时间
     * @param unit        时间单位
     * @param valueType   数据类型
     * @return
     * @author ebert_chan
     * @date 2023年5月22日
     */
    public <T, V> List<V> batchGetAndSet(Map<T, String> keys, Function<List<T>, Map<T, V>> valueMapper, long timeout, TimeUnit unit, Class<V> valueType) {
        // 定义一个数据接收对象
        List<V> result = new ArrayList<>(keys.size());

        // 未缓存的数据，key=T，value=缓存key
        Map<T, String> notExistsKey = new HashMap<>(keys.size());

        // 优先从缓存获取数据
        for (Map.Entry<T, String> entry : keys.entrySet()) {
            T key = entry.getKey();
            String value = entry.getValue();
            V t = this.getOrCombine(entry.getValue(), valueType, () -> notExistsKey.put(key, value));
            if (t == null) {
                // 能走到这里说明缓存命中，数据库查不到记录
                continue;
            }
            result.add(t);
        }

        // 未缓存数据，从外部查询数据
        List<T> ids = new ArrayList<>(notExistsKey.keySet());
        Map<T, V> queryResult = valueMapper.apply(ids);

        for (T id : ids) {
            V value = queryResult != null ? queryResult.get(id) : null;

            // 数据放入缓存
            this.set(notExistsKey.get(id), value, timeout, unit);

            if (value != null) {
                result.add(value);
            }
        }

        return result;
    }

    /**
     * 获取缓存的值，若缓存中无数据，执行callback逻辑
     * <p>
     *
     * @param key
     * @param valueType
     * @param callback
     * @return
     * @author ebert_chan
     * @date 2023年5月22日
     */
    @SuppressWarnings("unchecked")
    private <V> V getOrCombine(String key, Class<V> valueType, Runnable callback) {
        byte[] bytes = this.opsForValue().get(key);
        V value;
        if (bytes == null) {
            callback.run();
            return null;
        } else {
            if (bytes.length == 0) {
                return null;
            }
            value = this.deserializeForBytes(bytes, valueType);
        }
        return value;
    }

    private byte[] defaultEmptyBytes(String text) {
        return stringRedisSerializer.serialize(text == null ? "" : text);
    }

    private byte[] defaultEmptyBytes(byte[] bytes) {
        if (bytes == null) {
            return stringRedisSerializer.serialize("");
        }
        return bytes;
    }

    /**
     * 对象序列化为byte[]
     *
     * @param source
     * @return
     * @auther ebert_chan
     */
    private byte[] serializeToBytes(Object source) {
        if (source == null) {
            return null;
        }
        byte[] bytes;
        if (source instanceof byte[]) {
            return (byte[]) source;
        }
        if (source instanceof Iterable || source instanceof Map) {
            bytes = serializeForJdkSerializer(source);
        } else {
            String text = this.serializeToString(source);
            bytes = stringRedisSerializer.serialize(text);
        }
        return bytes;
    }

    /**
     * 对象序列化为string，基础数据类型返回字符串，否则返回JSON格式字符串
     *
     * @param source
     * @return
     * @auther ebert_chan
     */
    private String serializeToString(Object source) {
        if (source == null) {
            return null;
        }
        Class<?> clazz = source.getClass();
        boolean isBasicType = clazz.equals(String.class) || clazz.equals(Byte.class) || clazz.equals(Integer.class) || clazz.equals(Long.class) || clazz.equals(Double.class) || clazz.equals(Float.class) || clazz.equals(Short.class) || clazz.equals(Boolean.class) || clazz.equals(Character.class);

        if (isBasicType) {
            return String.valueOf(source);
        }

        if (clazz.equals(BigDecimal.class)) {
            return ((BigDecimal) source).toPlainString();
        }

        return JsonUtils.toJsonString(source);
    }

    /**
     * 通过JDK提供的序列化方式转化为byte[]
     *
     * @param source
     * @return
     * @auther ebert_chan
     */
    public byte[] serializeForJdkSerializer(Object source) {
        if (source == null) {
            return null;
        }
        return jdkRedisSerializer.serialize(source);
    }

    /**
     * 通过JDK提供的序列化方式将byte[]反序列化为指定类型对象
     *
     * @param bytes
     * @return
     * @auther ebert_chan
     */
    @SuppressWarnings("unchecked")
    public <T> T deserializeForJdkSerializer(byte[] bytes) {
        if (bytes == null) {
            return null;
        }
        return (T) jdkRedisSerializer.deserialize(bytes);
    }

    /**
     * byte[]反序列化为指定类型对象
     *
     * @param bytes
     * @param valueType
     * @return
     * @auther ebert_chan
     */
    private <T> T deserializeForBytes(byte[] bytes, Class<T> valueType) {
        if (bytes == null || bytes.length == 0 || valueType == null) {
            return null;
        }
        if (valueType.equals(byte[].class)) {
            return (T) bytes;
        }

        T value;
        if (Iterable.class.isAssignableFrom(valueType) || Map.class.isAssignableFrom(valueType)) {
            value = this.deserializeForJdkSerializer(bytes);
        } else {
            String text = stringRedisSerializer.deserialize(bytes);
            value = this.deserializeForStringValue(text, valueType);
        }
        return value;
    }

    /**
     * string反序列化指定类型对象。如果是基本数据类型，转换后返回，否则自动对text进行JSON反序列化操作（会根据JSON自动转换返回类型）
     *
     * @param text
     * @param valueType
     * @return
     * @auther ebert_chan
     */
    @SuppressWarnings("unchecked")
    private <T> T deserializeForStringValue(String text, Class<T> valueType) {
        if (text == null || valueType == null) {
            return null;
        }

        T value = deserializeToBasicTypeForStringValue(text, valueType);
        if (value != null) {
            return value;
        }

        Object object = JSON.parse(text);
        if (object instanceof JSONArray) {
            JSONArray jsonArray = (JSONArray) object;
            return (T) jsonArray.toJavaList(valueType);
        } else if (object instanceof JSONObject) {
            JSONObject jsonObject = (JSONObject) object;
            return jsonObject.toJavaObject(valueType);
        }

        return null;
    }

    /**
     * string反序列化指定类型对象。如果是基本数据类型，转换后返回，否则反序列化为指定类型引用的对象
     * <p>
     *
     * @param text
     * @param typeReference
     * @return T
     * @author ebert_chan
     * @date 2024/6/4
     */
    private <T> T deserializeForStringValue(String text, TypeReference<T> typeReference) {
        if (text == null || typeReference == null) {
            return null;
        }

        Type valueType = typeReference.getType();

        T value = deserializeToBasicTypeForStringValue(text, valueType);
        if (value != null) {
            return value;
        }

        return JsonUtils.parse(text, typeReference);
    }

    /**
     * string反序列化为基本数据类型
     * <p>
     *
     * @param text
     * @param valueType
     * @return T
     * @author ebert_chan
     * @date 2024/6/4
     */
    @SuppressWarnings("unchecked")
    private <T> T deserializeToBasicTypeForStringValue(String text, Type valueType) {
        if (text == null || valueType == null) {
            return null;
        }
        if (valueType.equals(String.class)) {
            return (T) text;
        }
        if (valueType.equals(Integer.class) || valueType.equals(Integer.TYPE)) {
            return (T) Integer.valueOf(text);
        }
        if (valueType.equals(Long.class) || valueType.equals(Long.TYPE)) {
            return (T) Long.valueOf(text);
        }
        if (valueType.equals(Byte.class) || valueType.equals(Byte.TYPE)) {
            return (T) Byte.valueOf(text);
        }
        if (valueType.equals(Double.class) || valueType.equals(Double.TYPE)) {
            return (T) Double.valueOf(text);
        }
        if (valueType.equals(Float.class) || valueType.equals(Float.TYPE)) {
            return (T) Float.valueOf(text);
        }
        if (valueType.equals(Short.class) || valueType.equals(Short.TYPE)) {
            return (T) Short.valueOf(text);
        }
        if (valueType.equals(Boolean.class) || valueType.equals(Boolean.TYPE)) {
            return (T) Boolean.valueOf(text);
        }
        if (valueType.equals(Character.class) || valueType.equals(Character.TYPE)) {
            return (T) Character.valueOf(text.charAt(0));
        }
        if (valueType.equals(BigDecimal.class)) {
            return (T) new BigDecimal(text);
        }
        return null;
    }

    /**
     * 转换key
     *
     * @param keyPattern
     * @param keyVariables
     * @return
     * @auther ebert_chan
     */
    public String covertKey(String keyPattern, Object... keyVariables) {
        Object[] arguments = keyVariables.clone();
        for (int i = 0; i < arguments.length; i++) {
            if (arguments[i] == null) {
                arguments[i] = "";
            } else {
                arguments[i] = String.valueOf(arguments[i]);
            }
        }
        return MessageFormat.format(keyPattern, arguments);
    }

    /*
     * @see java.io.Closeable#close()
     */
    @Override
    public void close() throws IOException {
        destoryFlag = true;
    }

}
